package com.example;

import cn.hutool.core.date.LocalDateTimeUtil;
import com.example.entity.Order;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;


/**
 * Author wangJinLong
 * Date 2025/8/25 17:35
 **/
public class FlinkCep {

    private static List<String> faceList = Arrays.asList("14901", "24902", "34903", "44903", "54903");

    public static void main(String[] args) throws Exception {
        nextPattern();
    }

    private static void simplePattern() throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStream<Order> customerDataStream = getOrderSource(environment);

        Pattern<Order, ?> pattern = Pattern.<Order>begin("first")
                .where(SimpleCondition.of(event -> event.getAmount() > 5))
                .times(3)
                .within(Duration.ofSeconds(5));

        PatternStream<Order> patternStream = CEP.pattern(customerDataStream.keyBy((KeySelector<Order, String>) Order::getOrderId), pattern);

        DataStream<String> result = patternStream.process(getProcessFunction("key stream pattern"));

        result.print();

        environment.execute();
    }

    private static void nextPattern() throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStream<Order> customerDataStream = getOrderSource(environment);

        // 连续三次 amound > 5
        Pattern<Order, ?> pattern = Pattern.<Order>begin("start")
                .where(SimpleCondition.of(e -> e.getAmount() > 5))
                .next("second")
                .where(SimpleCondition.of(e -> e.getAmount() > 5))
                .next("third")
                .where(SimpleCondition.of(e -> e.getAmount() > 5))
                .within(Duration.ofSeconds(5));

        PatternStream<Order> patternStream = CEP.pattern(customerDataStream, pattern);

        DataStream<String> result = patternStream.process(getProcessFunction("连续3次大于5"));

        result.print();

        environment.execute();
    }

    private static void condition2() throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStream<Order> customerDataStream = getOrderSource(environment);

        Pattern<Order, ?> pattern = Pattern.<Order>begin("first")
                .where(SimpleCondition.of(event -> Objects.equals(event.getOrderId(), "14901")))
                .where(SimpleCondition.of(event -> event.getAmount() > 5));

        PatternStream<Order> patternStream = CEP.pattern(customerDataStream, pattern);

        DataStream<String> result = patternStream.process(getProcessFunction("检测到 orderID：14901 并且 amound > 5"));

        result.print();

        environment.execute();
    }

    private static void simple() throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);

        DataStream<Order> customerDataStream = getOrderSource(environment);

        Pattern<Order, ?> pattern = Pattern.<Order>begin("first")
                .where(SimpleCondition.of(event -> Objects.equals(event.getOrderId(), "14901")))
//                .next("second")
//                .where(SimpleCondition.of(event -> event.getAmount() > 5));
                .next("second")
                .where(SimpleCondition.of(event -> Objects.equals(event.getOrderId(), "24902")));

        PatternStream<Order> patternStream = CEP.pattern(customerDataStream, pattern);

        DataStream<String> result = patternStream.process(getProcessFunction("检测到 orderID：14901 和 amound > 5 的"));

        result.print();

        environment.execute();
    }

    private static PatternProcessFunction<Order, String> getProcessFunction(String record) {
        return new PatternProcessFunction<>() {
            @Override
            public void processMatch(
                    Map<String, List<Order>> pattern,
                    Context ctx,
                    Collector<String> out
            ) {
                Set<Map.Entry<String, List<Order>>> entries = pattern.entrySet();
                for (Map.Entry<String, List<Order>> entry : entries) {
                    System.out.println(entry.getKey());
                    entry.getValue().forEach(order -> out.collect(order.toString()));
                }
                out.collect(record);
            }
        };
    }

    public static DataStream<Order> getOrderSource(StreamExecutionEnvironment env) {
        GeneratorFunction<Long, Order> generatorFunction = value -> {
            Order order = new Order();
            Random random = new Random();
            order.setOrderId(faceList.get(random.nextInt(faceList.size())));
            order.setAmount(random.nextInt(10) + 1);
            order.setTotalPrice(random.nextDouble(100));

            LocalDateTime now = LocalDateTime.now();
            order.setLocalOrderTime(LocalDateTimeUtil.format(now, "yyyy-MM-dd HH:mm:ss"));
            long epochMilli = now.toInstant(ZoneOffset.systemDefault().getRules().getOffset(now)).toEpochMilli();
            order.setOrderTime(epochMilli);

            return order;
        };

        DataGeneratorSource<Order> source = new DataGeneratorSource<>(
                generatorFunction,
                60L,
                RateLimiterStrategy.perSecond(1),
                Types.POJO(Order.class)
        );

        DataStream<Order> mapSource = env.fromSource(
                source,
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(((element, recordTimestamp) -> element.getOrderTime()))
                , "map source"
        );

        return mapSource;
    }

}
